package cgl.hadoopsensorgrid.sensorgrid.ryotoeval;

import java.lang.InterruptedException;
import java.lang.StringBuffer;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.io.IOException;
import java.io.File;
import java.io.FileWriter;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.net.ServerSocket;
import java.net.InetAddress;
import java.util.GregorianCalendar;
import java.util.Calendar;
import java.util.StringTokenizer;
import java.util.Iterator;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.text.SimpleDateFormat;

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;

import cgl.narada.util.ByteUtilities;

import cgl.sensorgrid.common.PropertyFile;
import cgl.sensorgrid.common.ITRFtoLatLon;

import cgl.hadoopsensorgrid.sensorgrid.util.*;

/***
 * 
 * The ryoMapper extends the hadoop Mapper class. It is deployed by the
 * ryoController. It gets data from the rtdMapper via a socket and passes it to
 * ryoReducer.
 * 
 * @author Chao Sun
 * 
 */
public class ryoMapper extends
		Mapper<Object, Text, serializedText, BytesWritable> {
	private Properties properties;
	// variables for RYO
	private String ryoAddr;
	private int ryoPort;
	private String ryoAddress;
	private Socket ryoSocket = null;
	private InetAddress ryoInetAddr = null;
	private ObjectInputStream ryoInStream = null;
	// variables for eval
	private String evalInputSourse;
	private String asciiFile;
	private String posFile;
	// linkedlist for keeping and sending data sets
	private List<byte[]> asciiList = Collections
			.synchronizedList(new LinkedList<byte[]>());
	private List<byte[]> posList = Collections
			.synchronizedList(new LinkedList<byte[]>());
	private Object asciiSyncObj = new Object();
	private Object posSyncObj = new Object();

	/***
	 * 
	 * Class for converting RYO data to ASCII.
	 * 
	 * Code from sensorgrid.sopac.gps.filter.ryoTOascii.java by Galip Aydin.
	 * 
	 * @author Chao Sun
	 * 
	 */
	class toAscii extends Thread {
		private String header = "";
		private String position = "";
		private String xyzVariance = "";
		private String troposphericCovariance = "";
		private String satelliteInfo = "";
		private String RYOmessage = "";
		int[] a = new int[1];
		private byte[] type;
		private int PID = 1;
		private int ID = 1;
		private Context context;

		public toAscii(Context con) {
			this.type = "ASCII".getBytes();
			this.context = con;
		}

		public void run() {
			try {
				while (true) {
					synchronized (asciiSyncObj) {
						if (asciiList.size() == 0) {
							asciiSyncObj.wait();
						}
						byte[] data2 = asciiList.remove(0);
						if (data2.length == 9 && (new String(data2)).equals("Terminate")) {
							break;
						}
						// process data
						if (data2 != null && data2.length > 0) {
							processRawMessage(data2, ID);
							ID++;
						}
					}
				}
				return;
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}

		public void processRawMessage(byte[] input, int dataID) {
			try {
				boolean validHeader = false;
				if (input.length >= 5) {
					validHeader = decodeHeader(input);
				}
				if (validHeader && input.length > 52) {
					decodePosition(input, 5);
				}

				String pos = this.getPosition();
				if (pos.endsWith("\n")) {
					pos = pos.substring(0, pos.length() - 1);
				}
				if (pos != null && pos.trim() != "" && pos.trim().length() > 0) {
					String asciiData = pos + "\nEOF\n";
					byte[] asciiBytes = asciiData.getBytes();
					serializedText key = new serializedText(PID, type, dataID);
					int len = asciiBytes.length;
					BytesWritable value = new BytesWritable();
					value.set(asciiBytes, 0, len);
					context.write(key, value);
				}
				this.position = "";
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}

		public String messageToGML(String message) {
			try {
				return "";
			} catch (Exception ex) {
				ex.printStackTrace();
				return null;
			}
		}

		public boolean decodeHeader(byte[] input) {
			try {
				boolean validHeader = false;
				cgl.narada.util.ByteUtilities b = new cgl.narada.util.ByteUtilities(a);
				int SyncMarker1 = b.getIntFromByte(input[0]);
				int SyncMarker2 = b.getIntFromByte(input[1]);
				if (Integer.toHexString(SyncMarker1).toUpperCase().equalsIgnoreCase(
						"9C")
						&& Integer.toHexString(SyncMarker2).toUpperCase().equals("A5")) {
					validHeader = true;
				}
				return validHeader;
			} catch (Exception ex) {
				ex.printStackTrace();
				return false;
			}
		}

		public String decodePosition(byte[] input, int offset) {
			try {
				String line = "";
				ByteUtilities b = new ByteUtilities(a);
				int i = offset;
				ByteBuffer bb = ByteBuffer.wrap(input);
				bb.order(ByteOrder.LITTLE_ENDIAN);
				short GPSWeek = bb.getShort(i);
				int GPSmsOfWeek = bb.getInt(i + 2);
				String date = getDate(Short.toString(GPSWeek), Integer
						.toString(GPSmsOfWeek));
				int SiteIndex = b.getIntFromByte(input[i + 6]);
				int SiteCount = b.getIntFromByte(input[i + 7]);
				String SiteId = Character.toString((char) (input[i + 8]))
						+ Character.toString((char) (input[i + 9]))
						+ Character.toString((char) (input[i + 10]))
						+ Character.toString((char) (input[i + 11]));
				line += "\t" + date;
				line += "\t" + SiteId;
				line += "\t" + SiteIndex;
				line += "\t" + SiteCount;
				double X = bb.getDouble((i + 14));
				double Y = bb.getDouble((i + 22));
				double Z = bb.getDouble((i + 30));
				double T = bb.getDouble((i + 38));
				line += "\t" + X;
				line += "\t" + Y;
				line += "\t" + Z;
				line += "\t" + T;
				double lat = 0.0;
				double lon = 0.0;
				double hi = 0.0;
				ITRFtoLatLon latlon = new ITRFtoLatLon();
				double[] llh = null;
				llh = latlon.GetLatLonHeightRadians(X, Y, Z);
				if (llh != null && llh.length == 3) {
					line += "\t" + llh[0];
					line += "\t" + llh[1];
					line += "\t" + llh[2];
				}
				String posQualityBytes = b.printByte(input[i + 46]);
				if (!posQualityBytes.equals("") && posQualityBytes.length() == 8) {
					String method = posQualityBytes.substring(3, 6);
					String calc = posQualityBytes.substring(6, 8);
					if (calc.equals("00")) {
						line += " Undetermined";
					} else if (calc.equals("01")) {
						line += " L1";
					} else if (calc.equals("10")) {
						line += " L1/L2";
					}
					if (method.equals("001")) {
						line += " Undetermined";
					} else if (method.equals("010")) {
						line += " Absolute";
					} else if (method.equals("011")) {
						line += " Relative";
					} else if (method.equals("100")) {
						line += " Phase";
					}
				}
				String flagString = b.printByte(input[i + 47]);
				boolean xyz = false;
				boolean tropo = false;
				boolean satellite = false;
				if (!flagString.equals("") && flagString.length() == 8) {
					if (flagString.charAt(7) == '1') {
						line += " XYZ";
						xyz = true;
					}
					if (flagString.charAt(6) == '1') {
						line += " Troposhpheric";
						tropo = true;
					}
					if (flagString.charAt(5) == '1') {
						line += " Satellite";
						satellite = true;
					}
				}
				i += 47;
				if (input.length > 51) {
					if (xyz == true) {
						i = decodeXYZVariances(input, i + 1);
					}
					line += " " + this.getXyzVariance();
					this.setXyzVariance("");
				}
				if (input.length > i) {
					if (satellite == true) {
						i = decodeSatellite(input, i + 1);
					}
					line += " " + this.getSatelliteInfo();
					this.setSatelliteInfo("");
				}
				if (line != null && !line.equals("")) {
					this.appendPosition(line.trim() + "\n");
				}
				if (input.length > i + 52) {
					decodePosition(input, i + 7);
				}
				return line;
			} catch (Exception ex) {
				System.err.println("\nException occured in decodePosition()");
				ex.printStackTrace();
				return null;
			}
		}

		public int decodeXYZVariances(byte[] input, int i) {
			try {
				String line = "";
				ByteBuffer bb = ByteBuffer.wrap(input);
				bb.order(ByteOrder.LITTLE_ENDIAN);
				double scale = bb.getDouble(i);
				double Xvar = bb.getDouble((i + 8));
				double Yvar = bb.getDouble((i + 16));
				double Zvar = bb.getDouble((i + 24));
				double YXvar = bb.getDouble((i + 32));
				double YZvar = bb.getDouble((i + 40));
				double ZXvar = bb.getDouble((i + 48));
				i += 55;
				line += " " + scale;
				line += " " + Xvar;
				line += " " + Yvar;
				line += " " + Zvar;
				line += " " + YXvar;
				line += " " + YZvar;
				line += " " + ZXvar;
				this.setXyzVariance(line);
				return i;
			} catch (Exception ex) {
				return i;
			}
		}

		public int decodeTroposphericCoverianceBlock(byte[] input, int i) {
			try {
				String line = "";
				ByteBuffer bb = ByteBuffer.wrap(input);
				bb.order(ByteOrder.LITTLE_ENDIAN);
				double Tvar = bb.getDouble((i + 8));
				double TXcovar = bb.getDouble((i + 16));
				double TYcovar = bb.getDouble((i + 24));
				double TZcovar = bb.getDouble((i + 32));
				i += 33;
				line += " " + Tvar;
				line += " " + TXcovar;
				line += " " + TYcovar;
				line += " " + TZcovar;
				this.setTroposphericCovariance(line);
				return i;
			} catch (Exception ex) {
				return i;
			}
		}

		public int decodeSatellite(byte[] input, int i) {
			try {
				String line = "";
				ByteUtilities b = new ByteUtilities(a);
				ByteBuffer bb = ByteBuffer.wrap(input);
				bb.order(ByteOrder.LITTLE_ENDIAN);
				int satelliteCount = b.getIntFromByte(input[i]);
				double GDOP = bb.getDouble((i + 1));
				if (satelliteCount > 0) {
					i += 9;
					for (int j = 0; j < satelliteCount; j++) {
						if (input.length > i + 5) {
							int PRN = b.getIntFromByte(input[i]);
							int PRNFlags = b.getIntFromByte(input[i + 1]);
							short Elevation = bb.getShort(i + 2);
							short Azimuth = bb.getShort(i + 4);
							line += " " + (j + 1);
							line += " " + PRN;
							line += " " + PRNFlags;
							line += " " + Elevation;
							line += " " + Azimuth;
							i += 6;
						}
					}
					this.setSatelliteInfo(line);
				}
				return i;
			} catch (Exception ex) {
				return i;
			}
		}

		public void setHeader(String header) {
			this.header = header;
		}

		public void setPosition(String position) {
			this.position = position;
		}

		public void appendPosition(String pos) {
			this.position += pos;
		}

		public void setXyzVariance(String xyzVariance) {
			this.xyzVariance = xyzVariance;
		}

		public void setTroposphericCovariance(String troposphericCovariance) {
			this.troposphericCovariance = troposphericCovariance;
		}

		public void setSatelliteInfo(String satelliteInfo) {
			this.satelliteInfo = satelliteInfo;
		}

		public void setRYOmessage(String header, String position) {
			String RYOmsg = header;
			RYOmsg.trim();
			RYOmsg += position;
			RYOmsg.trim();
			this.RYOmessage = RYOmsg;
		}

		public String getHeader() {
			return header;
		}

		public String getPosition() {
			return position;
		}

		public String getXyzVariance() {
			return xyzVariance;
		}

		public String getTroposphericCovariance() {
			return troposphericCovariance;
		}

		public String getSatelliteInfo() {
			return satelliteInfo;
		}

		public String getRYOmessage() {
			return RYOmessage;
		}

		public String getDate(String GPS_week, String GPS_millisecond) {
			try {
				long GPS_start_epoch = 315964800;
				long GPS_leapseconds = 13;
				long time = GPS_start_epoch + Long.parseLong(GPS_week) * 7 * 24 * 3600
						+ Long.parseLong(GPS_millisecond) / 1000 - GPS_leapseconds;
				Calendar cal = Calendar.getInstance();
				cal.setTimeInMillis(time * 1000);
				SimpleDateFormat myDateFormat = new SimpleDateFormat(
						"yyyy-MM-dd\thh:mm:ssa-z");
				return myDateFormat.format(cal.getTime());
			} catch (Exception ex) {
				ex.printStackTrace();
				return null;
			}
		}

	}

	/***
	 * 
	 * Class for converting RYO data to POS.
	 * 
	 * Code from sensorgrid.sopac.gps.filter.ryoTOpos.java by Galip Aydin.
	 * 
	 * @author Chao Sun
	 * 
	 */
	class toPos extends Thread {
		private String header = "";
		private String position = "";
		private String xyzVariance = "";
		private String troposphericCovariance = "";
		private String satelliteInfo = "";
		private String RYOmessage = "";
		int[] a = new int[1];
		private byte[] type;
		private int PID = 2;
		private int ID = 1;
		private Context context;

		public toPos(Context con) {
			this.type = "POS".getBytes();
			this.context = con;
		}

		/**
		 * Retrieves RYO data from list.
		 */
		public void run() {
			try {
				while (true) {
					synchronized (posSyncObj) {
						if (posList.size() == 0) {
							posSyncObj.wait();
						}
						byte[] data2 = posList.remove(0);
						int dataLen = data2.length;
						// System.out.println("toPos filter retrieved data");
						if (dataLen == 9 && (new String(data2)).equals("Terminate")) {
							break;
						}
						// process data
						if (data2 != null && dataLen > 0) {
							processRawMessage(data2, ID);
							ID++;
						}
					}
				}
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}

		/**
		 * Adds POS data to list.
		 * 
		 * @param input
		 *          byte[]
		 */
		public void processRawMessage(byte[] input, int datID) {
			try {
				boolean validHeader = false;
				if (input.length >= 5) {
					validHeader = decodeHeader(input);
				}
				if (validHeader && input.length > 52) {
					decodePosition(input, 5);
				}

				String pos = this.getPosition();
				if (pos.endsWith("\n")) {
					pos = pos.substring(0, pos.length() - 1);
				}
				// System.out.println("POS Data: " + pos);
				if (pos != null && pos.trim() != "" && pos.trim().length() > 0) {
					String posData = pos + "\nEOF\n";
					byte[] posBytes = posData.getBytes();
					serializedText key = new serializedText(PID, type, datID);
					int len = posBytes.length;
					BytesWritable value = new BytesWritable();
					value.set(posBytes, 0, len);
					context.write(key, value);
					// System.out.println("toPos filter writes data");
				}
				this.position = "";
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}

		public String messageToGML(String message) {
			try {
				return "";
			} catch (Exception ex) {
				ex.printStackTrace();
				return null;
			}
		}

		public boolean decodeHeader(byte[] input) {
			try {
				boolean validHeader = false;
				cgl.narada.util.ByteUtilities b = new cgl.narada.util.ByteUtilities(a);
				int SyncMarker1 = b.getIntFromByte(input[0]);
				int SyncMarker2 = b.getIntFromByte(input[1]);
				if (Integer.toHexString(SyncMarker1).toUpperCase().equalsIgnoreCase(
						"9C")
						&& Integer.toHexString(SyncMarker2).toUpperCase().equals("A5")) {
					validHeader = true;
				}
				return validHeader;
			} catch (Exception ex) {
				ex.printStackTrace();
				return false;
			}
		}

		public String decodePosition(byte[] input, int offset) {
			try {
				String line = "";
				ByteUtilities b = new ByteUtilities(a);
				int i = offset;
				ByteBuffer bb = ByteBuffer.wrap(input);
				bb.order(ByteOrder.LITTLE_ENDIAN);
				short GPSWeek = bb.getShort(i);
				int GPSmsOfWeek = bb.getInt(i + 2);
				String date = getDate(Short.toString(GPSWeek), Integer
						.toString(GPSmsOfWeek));
				int SiteIndex = b.getIntFromByte(input[i + 6]);
				int SiteCount = b.getIntFromByte(input[i + 7]);
				String SiteId = Character.toString((char) (input[i + 8]))
						+ Character.toString((char) (input[i + 9]))
						+ Character.toString((char) (input[i + 10]))
						+ Character.toString((char) (input[i + 11]));
				line += "\t" + SiteId;
				line += "\t" + date;
				double X = bb.getDouble((i + 14));
				double Y = bb.getDouble((i + 22));
				double Z = bb.getDouble((i + 30));
				double T = bb.getDouble((i + 38));
				line += "\t" + X;
				line += "\t" + Y;
				line += "\t" + Z;
				line += "\t" + T;
				double lat = 0.0;
				double lon = 0.0;
				double hi = 0.0;
				ITRFtoLatLon latlon = new ITRFtoLatLon();
				double[] llh = null;
				llh = latlon.GetLatLonHeightRadians(X, Y, Z);
				if (llh != null && llh.length == 3) {
					line += "\t" + llh[0];
					line += "\t" + llh[1];
					line += "\t" + llh[2];
				}
				if (input.length > i + 47) {
					String flagString = b.printByte(input[i + 47]);
					boolean xyz = false;
					boolean tropo = false;
					boolean satellite = false;
					if (!flagString.equals("") && flagString.length() == 8) {
						if (flagString.charAt(7) == '1') {
							xyz = true;
						}
						if (flagString.charAt(6) == '1') {
							tropo = true;
						}
						if (flagString.charAt(5) == '1') {
							satellite = true;
						}
					}
					i += 47;
					if (input.length > 51) {
						if (xyz == true) {
							i = decodeXYZVariances(input, i + 1);
						}
						line += " " + this.getXyzVariance();
						this.setXyzVariance("");
					}
					if (input.length > i) {
						if (satellite == true) {
							i = decodeSatellite(input, i + 1);
						}
						line += " " + this.getSatelliteInfo();
						this.setSatelliteInfo("");
					}
					if (line != null && !line.equals("")) {
						this.appendPosition(line.trim() + "\n");
					}
					if (input.length > i + 52) {
						decodePosition(input, i + 7);
					}
				}
				return line;
			} catch (Exception ex) {
				System.err.println("\nException occured in decodePosition()");
				ex.printStackTrace();
				return null;
			}
		}

		public int decodeXYZVariances(byte[] input, int i) {
			try {
				String line = "";
				ByteBuffer bb = ByteBuffer.wrap(input);
				bb.order(ByteOrder.LITTLE_ENDIAN);
				i += 55;
				return i;
			} catch (Exception ex) {
				return i;
			}
		}

		public int decodeTroposphericCoverianceBlock(byte[] input, int i) {
			try {
				String line = "";
				ByteBuffer bb = ByteBuffer.wrap(input);
				bb.order(ByteOrder.LITTLE_ENDIAN);
				i += 33;
				this.setTroposphericCovariance(line);
				return i;
			} catch (Exception ex) {
				return i;
			}
		}

		public int decodeSatellite(byte[] input, int i) {
			try {
				String line = "";
				ByteUtilities b = new ByteUtilities(a);
				ByteBuffer bb = ByteBuffer.wrap(input);
				bb.order(ByteOrder.LITTLE_ENDIAN);
				int satelliteCount = b.getIntFromByte(input[i]);
				double GDOP = bb.getDouble((i + 1));
				if (satelliteCount > 0) {
					i += 9;
					for (int j = 0; j < satelliteCount; j++) {
						if (input.length > i + 5) {
							i += 6;
						}
					}
					this.setSatelliteInfo(line);
				}
				return i;
			} catch (Exception ex) {
				return i;
			}
		}

		public void setHeader(String header) {
			this.header = header;
		}

		public void setPosition(String position) {
			this.position = position;
		}

		public void appendPosition(String pos) {
			this.position += pos;
		}

		public void setXyzVariance(String xyzVariance) {
			this.xyzVariance = xyzVariance;
		}

		public void setTroposphericCovariance(String troposphericCovariance) {
			this.troposphericCovariance = troposphericCovariance;
		}

		public void setSatelliteInfo(String satelliteInfo) {
			this.satelliteInfo = satelliteInfo;
		}

		public void setRYOmessage(String header, String position) {
			String RYOmsg = header;
			RYOmsg.trim();
			RYOmsg += position;
			RYOmsg.trim();
			this.RYOmessage = RYOmsg;
		}

		public String getHeader() {
			return header;
		}

		public String getPosition() {
			return position;
		}

		public String getXyzVariance() {
			return xyzVariance;
		}

		public String getTroposphericCovariance() {
			return troposphericCovariance;
		}

		public String getSatelliteInfo() {
			return satelliteInfo;
		}

		public String getRYOmessage() {
			return RYOmessage;
		}

		public String getDate(String GPS_week, String GPS_millisecond) {
			try {
				long GPS_start_epoch = 315964800;
				long GPS_leapseconds = 13;
				long time = GPS_start_epoch + Long.parseLong(GPS_week) * 7 * 24 * 3600
						+ Long.parseLong(GPS_millisecond) / 1000 - GPS_leapseconds;
				Calendar cal = Calendar.getInstance();
				cal.setTimeInMillis(time * 1000);
				SimpleDateFormat myDateFormat = new SimpleDateFormat(
						"yyyy-MM-dd\thh:mm:ssa-z");
				return myDateFormat.format(cal.getTime());
			} catch (Exception ex) {
				ex.printStackTrace();
				return null;
			}
		}

	}

	/***
	 * 
	 * Class for connecting to RYO sourse, getting RYO data and adding data sets
	 * to the datalists. It also initializes filters for multiple types of
	 * objective data types and starts the filters.
	 * 
	 * @author Chao Sun
	 * 
	 */
	class ryoToEval {
		private toAscii toa;
		private toPos top;
		private Context context;

		/**
		 * Construnctor. Get context.
		 * 
		 * @param context
		 *          context of this ryoMapper object
		 */
		public ryoToEval(Context con) {
			this.context = con;
		}

		/**
		 * This method connects to an RYO socket address.
		 */
		private void connectRyoSocket() {
			try {
				ryoSocket = new Socket(ryoInetAddr, ryoPort);
				ryoInStream = new ObjectInputStream(ryoSocket.getInputStream());
			} catch (IOException ioE) {
				System.out.println(ryoInetAddr.toString() + ":" + ryoPort
						+ " RYO Socket connection unsuccessful. Terminating...");
				System.out.println(ioE.toString());
				System.exit(0);
			}
		}

		/**
		 * Add RYO data to ASCII list
		 */
		private void addAsciiData(byte[] data) {
			synchronized (asciiSyncObj) {
				asciiList.add(data);
				if (asciiList.size() == 1) {
					asciiSyncObj.notify();
				}
			}
		}

		/**
		 * Add RYO data to POS list
		 */
		private void addPosData(byte[] data) {
			synchronized (posSyncObj) {
				posList.add(data);
				if (posList.size() == 1) {
					posSyncObj.notify();
				}
			}
		}

		/**
		 * Initialize toAscii and toPos filters and start them. On reveiving a data
		 * tuple, add them to the list of the filters.
		 * 
		 * @throws InterruptedException
		 */
		public void run() throws InterruptedException {
			// connect to the RYO sourse
			connectRyoSocket();
			System.out.println("Connection successful to RYO source " + ryoAddress);
			// initialize filters and start
			toAscii toa = new toAscii(context);
			toPos top = new toPos(context);
			toa.start();
			top.start();

			while (true) {
				try {
					byte[] ryoData = (byte[]) (ryoInStream.readObject());
					int len = ryoData.length;
					byte[] backupRyoData = new byte[len];
					System.arraycopy(ryoData, 0, backupRyoData, 0, len);

					// add data to lists System.out.println("get RYO data");
					addAsciiData(ryoData);
					Thread.sleep(10);
					addPosData(backupRyoData);
					Thread.sleep(10);

					// TIMEOUT reached. Return
					if (len == 9 && (new String(ryoData)).equals("Terminate")) {
						// System.out.println("terminating signal received from RTD");
						if (toa != null && toa.isAlive()) {
							System.out.println("ASCII filter still running");
							toa.join();
						}
						if (top != null && top.isAlive()) {
							System.out.println("POS filter still running");
							top.join();
						}
						// System.out.println("filters finished");
						break;
					}
				} catch (Exception e) {
					System.err.println("RYO data receiving error");
					e.printStackTrace();
				}
			} // while
		}

	}
	
	/**
	 * Override the setup method.
	 * 
	 * @param context
	 *          context of this ryoMapper object
	 */
	protected void setup(Context context) {
		loadProperties();
		System.out.println("Ryo mapper started");
	}

	/**
	 * This method overrides the map() method of class Mapper. It gets the RYO
	 * data sourse address and port number from a file and assigns them to the two
	 * global variables, and then starts the runRtdClient.
	 * 
	 * @param key
	 *          key of the data split
	 * @param value
	 *          contents of the data split, which is the address and port
	 * @param context
	 *          context of this ryoMapper object
	 * 
	 * @throws InterruptedException
	 * 
	 * @throws IOException
	 */
	protected void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {
		// value is the RYO sourse <Addr>:<Port>
		StringTokenizer sTokenizer = new StringTokenizer(value.toString(), ":");
		int tokenNum = sTokenizer.countTokens();
		if (tokenNum != 2) {
			System.err.println("Sourse list error. Format: " + "<Source IP>:<Port>");
			System.exit(1);
		}

		// parse the RYO sourse for IP address and port
		ryoAddr = sTokenizer.nextToken();
		ryoPort = Integer.parseInt(sTokenizer.nextToken());
		ryoAddress = ryoAddr + ":" + ryoPort;

		// parse the RYO IP address for ryoInetAddr
		StringTokenizer ipTokenizer = new StringTokenizer(ryoAddr, ".");
		byte[] ryoIP = new byte[4];
		int i = 0;
		while (ipTokenizer.hasMoreTokens()) {
			String number = ipTokenizer.nextToken();
			Integer num = new Integer(number);
			ryoIP[i++] = num.byteValue();
		}
		ryoInetAddr = InetAddress.getByAddress(ryoIP);

		// write temp file paths to file
		FileWriter filewriter = new FileWriter(evalInputSourse);
		filewriter.write(asciiFile + "\n" + posFile);
		filewriter.flush();
		filewriter.close();

		ryoToEval toEval = new ryoToEval(context);
		toEval.run();
		System.out.println("RYO mapper terminate");
	}
	
	/**
	 * Load properties from a given file.
	 */
	private void loadProperties() {
		try {
			properties = PropertyFile.loadProperties("hadoopsensorgrid.properties");
			String evalInputBase = properties.getProperty("hadoop.eval.inputbase");
			if (properties == null || evalInputBase == null) {
				System.err.println("Error: evalMapper failed to load properties!");
				System.exit(-1);
			}
			File fileDir = new File(evalInputBase);
			if (!fileDir.exists()) {
				fileDir.mkdirs();
			}
			evalInputSourse = evalInputBase + File.separator + "datalist.txt";
			asciiFile = evalInputBase + File.separator + "ascii.txt";
			posFile = evalInputBase + File.separator + "pos.txt";
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}